"""
NBA联盟赛季场均篮板分析应用（并行计算优化版本）

功能描述：
    使用PySpark对NBA比赛数据进行并行批处理分析，计算每个赛季的联盟场均篮板数。
    通过分区策略和并行计算提高处理效率。

优化策略：
    1. 按赛季ID进行分区，实现数据的并行处理
    2. 使用缓存机制减少重复计算
    3. 实现分布式计算，提高处理效率
"""

from pyspark.sql import SparkSession
from pyspark.sql.functions import sum, count, round, col
import os

def calculate_season_avg_rebounds_parallel():
    """计算NBA每个赛季的联盟场均篮板数（并行优化版本）
    
    优化要点：
    1. 设置合适的分区数
    2. 使用缓存减少重复计算
    3. 并行处理各赛季数据
    4. 合理使用shuffle操作
    """
    
    # 第一步：配置Spark环境(设置分区数和并行数为4)
    spark = SparkSession.builder \
        .appName("NBA Season Average Rebounds Parallel") \
        .config("spark.sql.shuffle.partitions", 4) \
        .config("spark.default.parallelism", 4) \
        .getOrCreate()

    # 设置日志级别为ERROR，减少不必要的日志输出
    spark.sparkContext.setLogLevel("ERROR")

    try:
        # 第二步：数据加载和分区优化
        csv_path = "../data/game_processed.csv"
        
        # 读取CSV文件并进行分区优化
        df = spark.read \
            .option("header", "true") \
            .option("inferSchema", "true") \
            .csv(csv_path) \
            .repartition(4, "season_id")  # 按season_id列重新分区，提高并行度
        
        # 缓存数据，避免重复计算
        df.cache()

        # 第三步：并行数据聚合
        # 按赛季分组并计算统计数据
        season_stats = df.groupBy("season_id").agg(
            sum("reb_home").alias("total_home_rebounds"),    # 计算主场篮板总数
            sum("reb_away").alias("total_away_rebounds"),    # 计算客场篮板总数
            count("reb_home").alias("num_games")             # 计算比赛场次
        )

        # 第四步：计算场均篮板数
        result = season_stats.select(
            "season_id",  # 选择赛季ID列
            round(  
                (col("total_home_rebounds") + col("total_away_rebounds")) / (col("num_games") * 2),
                2
            ).alias("avg_rebounds_per_team")  # 将计算结果列命名为avg_rebounds_per_team
        ).orderBy("season_id")  # 按赛季ID排序

        # 第五步：结果输出优化
        output_path = "../data/season_avg_rebounds_parallel.csv"
        result.coalesce(1).write \
            .mode("overwrite") \
            .option("header", "true") \
            .csv(output_path)

        # 第六步：显示结果和性能统计
        print("\n每个赛季的联盟场均篮板数：")
        result.show(truncate=False)

        # 显示执行计划，用于性能分析
        print("\n执行计划：")
        result.explain()

        # 第七步：计算和显示篮板数变化趋势
        first_season = result.orderBy("season_id").first()
        last_season = result.orderBy("season_id", ascending=False).first()
        
        if first_season and last_season:
            change = last_season["avg_rebounds_per_team"] - first_season["avg_rebounds_per_team"]
            change_percentage = (change / first_season["avg_rebounds_per_team"]) * 100
            
            print(f"\n篮板数演变趋势分析：")
            print(f"从{first_season['season_id']}赛季到{last_season['season_id']}赛季：")
            print(f"场均篮板数变化: {change:.2f}")
            print(f"变化百分比: {change_percentage:.2f}%")

            # 补充分析说明
            if change > 0:
                print("篮板数呈上升趋势，可能反映出比赛节奏加快或投篮命中率变化")
            elif change < 0:
                print("篮板数呈下降趋势，可能反映出投篮命中率提高或比赛风格变化")
            else:
                print("篮板数基本保持稳定")

        # 第八步：清理缓存
        df.unpersist()

    finally:
        # 第九步：清理资源
        spark.stop()

if __name__ == "__main__":
    calculate_season_avg_rebounds_parallel()
